package com.flute.icrawler.framework.executor;

import org.apache.log4j.Logger;

import com.flute.icrawler.app.processor.AbstractProcessor;
import com.flute.icrawler.framework.framework.CrawlUrl;
import com.flute.icrawler.framework.framework.LocalCrawlUrlPool;

//import com.cpkf.yyjd.tools.util.collection.Accumulator;

/**
 * 多线程的处理器装饰器，传入processor，使之具有多线程的能力
 * 
 * @author Administrator
 * 
 */
public class MultiThreadProcessor extends Thread {
	private static Logger logger = Logger.getLogger(MultiThreadProcessor.class);
	private AbstractProcessor processor;
	private LocalCrawlUrlPool pool;
	private ProcessorExecutor executor;
	private volatile boolean finish;
	public static final int SLEEP_INTERVAL = 10;

	// private static Accumulator<String> processAccumulator = new
	// Accumulator<String>();
	// private static Accumulator<String> freeAccumulator = new
	// Accumulator<String>();
	// private IntervalExecutor logExecutor;

	// private static long maxTime = 0;

	/**
	 * 每一个线程实例绑定一种类型的IProcessor
	 */
	public MultiThreadProcessor(AbstractProcessor processor,
			LocalCrawlUrlPool pool, ProcessorExecutor executor) {
		this.processor = processor;
		this.pool = pool;
		this.executor = executor;
		// this.logExecutor = getIntervalExecutor();
		// logExecutor.setInterval(100000);
		// logExecutor.start();
		super.setName(processor.getClass().getSimpleName()
				+ System.currentTimeMillis());
	}

	public void setFinish(boolean finish) {
		this.finish = finish;
	}

	@Override
	public void run() {
		while (!finish) {
			// synchronized (processAccumulator) {
			// processAccumulator.addKey(processor.getClass().getSimpleName()
			// + Thread.currentThread().getId());
			// }
			// long s = System.currentTimeMillis();

			// 向pool询问有是否有url满足了被当前这种类型的processor处理

			CrawlUrl url = pool.next(processor);

			// if (processor.getClass().equals(Print4Processor.class)) {
			// System.out
			// .println("time 1:" + (System.currentTimeMillis() - s));
			// }
			// long time = System.currentTimeMillis() - s;
			// if (time > maxTime) {
			// maxTime = time;
			// }
			// System.out.println("maxtime:" + maxTime);
			if (url != null) {
				try {
					// 标记为正在被该类型的processor处理
					url.processing(processor);
				//	long s = System.currentTimeMillis();
					processor.process(url);
				//	logger.info(processor + " process time:"
						//	+ (System.currentTimeMillis() - s));
				} catch (Exception e) {
					logger.error("\nexception in processor '"
							+ processor.getClass().getSimpleName() + ":"
							+ url.getUrl(), e);
					continue;
				}
				// if (processor.getClass().equals(Print4Processor.class)) {
				// System.out.println("time 2:"
				// + (System.currentTimeMillis() - s));
				// }
				url.processed(processor);
				// if (processor.getClass().equals(Print4Processor.class)) {
				// System.out.println("time 5:"
				// + (System.currentTimeMillis() - s));
				// }
				long time = System.currentTimeMillis();
				pool.check(url);
				logger.info(processor + " check time:"
						+ (System.currentTimeMillis() - time) + "ms");
				// if (processor.getClass().equals(Print4Processor.class)) {
				// System.out.println("time 3:"
				// + (System.currentTimeMillis() - s));
				// }
			} else {
				// synchronized (freeAccumulator) {
				// freeAccumulator.addKey(processor.getClass().getSimpleName()
				// + Thread.currentThread().getId());
				// }
				try {
					pool.recordFreeProcessor(processor);
					Thread.sleep(SLEEP_INTERVAL);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			// if (processor.getClass().equals(Print4Processor.class)) {
			// System.out
			// .println("time 4:" + (System.currentTimeMillis() - s));
			// }
		}
		// synchronized (freeAccumulator) {
		// freeAccumulator.delete(processor.getClass().getSimpleName()
		// + Thread.currentThread().getId());
		// }
		// synchronized (processAccumulator) {
		// processAccumulator.delete(processor.getClass().getSimpleName()
		// + Thread.currentThread().getId());
		// }
		executor.removeDeadThread(Thread.currentThread().getId());
		logger.info("Thread '" + processor.getClass().getSimpleName() + "-"
				+ Thread.currentThread().getId() + "' finished.");
	}
	// private IntervalExecutor getIntervalExecutor() {
	// return new IntervalExecutor() {
	// private Accumulator<String> lastFreeAccumulator = new
	// Accumulator<String>();
	// private Accumulator<String> lastProcessAccumulator = new
	// Accumulator<String>();
	//
	// @Override
	// public void execute() {
	// logger.info("new record:");
	// synchronized (freeAccumulator) {
	// for (String key : freeAccumulator.getKeys()) {
	// logger
	// .info("free:"
	// + key
	// + ": "
	// + (freeAccumulator.getCount(key) - lastFreeAccumulator
	// .getCount(key)));
	// }
	// }
	//				
	// synchronized (processAccumulator) {
	// for (String key : processAccumulator.getKeys()) {
	// logger
	// .info("process:"
	// + key
	// + ": "
	// + (processAccumulator.getCount(key) - lastProcessAccumulator
	// .getCount(key)));
	// }
	// }
	//				
	// synchronized (freeAccumulator) {
	// lastFreeAccumulator = freeAccumulator.clone();
	// }
	// synchronized (processAccumulator) {
	// lastProcessAccumulator = processAccumulator.clone();
	// }
	// }
	//
	// };
	// }
}
